#include "bthreadpool.h"

using namespace bhttp;

bthreadpool::bthreadpool(const std::string& name)
	: name_(name)
{
}

bthreadpool::~bthreadpool()
{
	stop();
}

void bthreadpool::start(int num_thd)
{
	if (running_) return;
	running_ = true;
	threads_.reserve(num_thd);
	for (int i = 0; i < num_thd; i++)
	{
		auto thd = std::make_shared<bthread>(std::bind(&bthreadpool::thread_func, this));
		thd->start();
		threads_.emplace_back(thd);
	}
}

void bthreadpool::stop()
{
	if (!running_) return;
	{
		std::unique_lock<std::mutex> lk(mutex_);
		running_ = false;
		not_empty_.notify_all();
	}
	for (auto& th : threads_)
		th->join();
}

void bthreadpool::run(const btask& tk)
{
	if (threads_.empty())
	{
		tk();
	}
	else
	{
		std::unique_lock<std::mutex> lk(mutex_);
		while (is_full())
		{
			not_full_.wait(lk);
		}
		queue_.emplace_back(tk);
		not_empty_.notify_one();
	}
}

void bthreadpool::run(btask&& tk)
{
	if (threads_.empty())
	{
		tk();
	}
	else
	{
		std::unique_lock<std::mutex> lk(mutex_);
		while (is_full())
			not_full_.wait(lk);
		queue_.emplace_back(std::move(tk));
		not_empty_.notify_one();
	}
}

void bthreadpool::thread_func()
{
	try {
		while (running_)
		{
			btask tk = take();
			if (tk) tk();
		}
	}
	catch (const std::exception& ex)
	{
		fprintf(stderr, "exception caught in bthreadpool %s\n", name_.c_str());
		fprintf(stderr, "reason: %s\n", ex.what());
		abort();
	}
}

bool bthreadpool::is_full() const
{
	return max_queue_size_ > 0 && queue_.size() >= max_queue_size_;
}

bthreadpool::btask bthreadpool::take()
{
	std::unique_lock<std::mutex> lk(mutex_);
	while (queue_.empty() && running_)
		not_empty_.wait(lk);
	btask tk;
	if (!queue_.empty())
	{
		tk = queue_.front();
		queue_.pop_front();
		if (max_queue_size_ > 0)
			not_full_.notify_one();
	}
	return tk;
}
